package team.bluepen.supermarket.data.entity;

import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import space.lingu.light.DataColumn;
import space.lingu.light.DataTable;
import space.lingu.light.PrimaryKey;
import team.bluepen.supermarket.kafka.KafkaDeserializer;
import team.bluepen.supermarket.kafka.KafkaSerializer;

import java.beans.ConstructorProperties;

/**
 * @author RollW
 */
@DataTable(name = "product_id_status_amount")
public class ProductIdStatusAmount {
    @JsonProperty("id")
    @DataColumn(name = "id")
    @PrimaryKey
    private final long id;

    @JsonProperty("activeCount")
    @DataColumn(name = "active_count")
    private final int activeCount;

    @JsonProperty("inactiveCount")
    @DataColumn(name = "inactive_count")
    private final int inactiveCount;

    @ConstructorProperties({"id", "activeCount", "inactiveCount"})
    public ProductIdStatusAmount(long id, int activeCount, int inactiveCount) {
        this.id = id;
        this.activeCount = activeCount;
        this.inactiveCount = inactiveCount;
    }

    public long getId() {
        return id;
    }

    public int getActiveCount() {
        return activeCount;
    }

    public int getInactiveCount() {
        return inactiveCount;
    }

    public static ProductIdStatusAmount add(ProductIdStatusAmount a, ProductIdStatusAmount b) {
        return new ProductIdStatusAmount(a.getId(),
                a.getActiveCount() + b.getActiveCount(),
                a.getInactiveCount() + b.getInactiveCount());
    }

    public static ProductIdStatusAmount plus(ProductIdStatusAmount a, ProductIdStatusPair b) {
        return new ProductIdStatusAmount(a.getId(),
                a.getActiveCount() + (b.isState() ? b.getAmount() : 0),
                a.getInactiveCount() + (b.isState() ? 0 : b.getAmount()));
    }

    public static ProductIdStatusAmount of(Product product) {
        return new ProductIdStatusAmount(product.getId(),
                product.isState() ? product.getAmount() : 0,
                product.isState() ? 0 : product.getAmount());
    }

    public static ProductIdStatusAmount of(ProductIdStatusPair pair) {
        return new ProductIdStatusAmount(pair.getId(),
                pair.isState() ? pair.getAmount() : 0,
                pair.isState() ? 0 : pair.getAmount());
    }

    public static KafkaSerializer<ProductIdStatusAmount> serializer() {
        return Serial.INSTANCE;
    }

    public static KafkaDeserializer<ProductIdStatusAmount> deserializer() {
        return Serial.INSTANCE;
    }

    private static class Serial implements KafkaSerializer<ProductIdStatusAmount>,
            KafkaDeserializer<ProductIdStatusAmount> {
        private final ObjectMapper objectMapper = new ObjectMapper();

        @Override
        public String serialize(ProductIdStatusAmount productIdStatusAmount) {
            try {
                return objectMapper.writeValueAsString(productIdStatusAmount);
            } catch (JsonProcessingException e) {
                throw new RuntimeException(e);
            }
        }

        @Override
        public ProductIdStatusAmount deserialize(String s) {
            try {
                return objectMapper.readValue(s, ProductIdStatusAmount.class);
            } catch (JsonProcessingException e) {
                throw new RuntimeException(e);
            }
        }

        static final Serial INSTANCE = new Serial();
    }
}
